package sender

import (
	"fmt"
	"github.com/IBM/sarama"
	"log"
)

type KafkaClient struct {
	IP    string
	Port  string
	Topic string
	sarama.SyncProducer
}

func (k *KafkaClient) Send(contents []byte) (int64, error) {
	msg := &sarama.ProducerMessage{}
	msg.Topic = k.Topic
	msg.Value = sarama.StringEncoder(contents)
	part, offset, err := k.SendMessage(msg)
	if err != nil {
		return 0, err
	}
	log.Printf("成功发送信息到 Topic: %s, Partition: %d, Offset: %d\n",
		k.Topic, part, offset)
	return int64(len(contents)), nil
}

func (k *KafkaClient) Close() error {
	return k.SyncProducer.Close()
}

// NewKafka 创建Sender的kafka发送端
func NewKafka(ip, port, topic string) (SenderCloser, error) {
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	config.Producer.Return.Successes = true

	producer, err := sarama.NewSyncProducer([]string{
		fmt.Sprintf("%s:%s", ip, port),
	}, config)
	if err != nil {
		log.Printf("创建Kafka客户端失败, 错误信息: %s\n",
			err.Error())
		return nil, err
	}
	return &KafkaClient{
		IP:           ip,
		Port:         port,
		Topic:        topic,
		SyncProducer: producer,
	}, nil
}

// 编译时检查KafkaClient是否实现了SenderCloser接口
var _ SenderCloser = &KafkaClient{}

func init() {
	// 包初始化, 准备默认值
}
